package com.zhang;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class OptimizedMultiThreadConsumer {

    private static final int BATCH_SIZE = 1000;
    private static final int QUEUE_CAPACITY = 10000; // 队列最大容量
    private BlockingQueue<String> queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY); // 阻塞队列

    private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); // 定时任务线程池

    public static void main(String[] args) {
        OptimizedMultiThreadConsumer consumer = new OptimizedMultiThreadConsumer();
        consumer.startConsuming();
        consumer.schedulePeriodicFlush();
    }

    // 启动多线程消费
    public void startConsuming() {
        ExecutorService executor = Executors.newFixedThreadPool(5); // 创建5个线程池

        for (int i = 0; i < 5; i++) {
            executor.submit(() -> {
                while (true) {
                    try {
                        String data = queue.take(); // 从队列中获取数据
                        processData(data); // 处理数据
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
            });
        }

        // 模拟生产者线程持续生产数据
        new Thread(() -> {
            while (true) {
                try {
                    String data = generateData();
                    queue.put(data); // 将数据放入队列
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }).start();

        executor.shutdown();
    }

    // 生成假数据
    private String generateData() {
        return "data-" + Math.random();
    }

    // 定时任务，每隔 10 秒检查队列并入库
    public void schedulePeriodicFlush() {
        scheduler.scheduleAtFixedRate(() -> {
            try {
                List<String> batchData = new ArrayList<>();
                queue.drainTo(batchData); // 将队列中的所有数据取出
                if (!batchData.isEmpty()) {
                    saveToDatabase(batchData);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, 10, 10, TimeUnit.SECONDS); // 每隔 10 秒执行一次
    }

    // 消费数据并批量入库
    public void processData(String data) {
        List<String> batchData = new ArrayList<>();
        queue.drainTo(batchData, BATCH_SIZE); // 获取 BATCH_SIZE 数量的数据
        if (!batchData.isEmpty()) {
            saveToDatabase(batchData);
        }
    }

    // 模拟数据批量入库
    private void saveToDatabase(List<String> batchData) {
        // 假设这里是数据库批量插入的操作
        System.out.println("Saving " + batchData);
    }
}